package org.example.four;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import org.example.utils.RabbitMqUtil;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 发布确认模式 比较三种方式
 * 1.单个确认
 * 2.批量确认
 * 3.异步批量确认,通过rabbitmq异步回调通知,哪些消息成功了,哪些失败了,发布者只管发送消息,用两个回调函数处理rabbitmq的回调就是
 */
public class ConfirmSelect {
    //发布消息数量
    public static final int NUM = 1000;
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //调用三种方法比较耗时
        //1.单个确认: 同步等待确认,简单,但吞吐量非常有限
        //single(); //当消息数量为: 1000 时,单个确认耗时: 118387
        //2.批量确认: 批量同步等待确认,简单,合理的吞吐量,但一旦出现问题很难推断出是哪条消息出现了问题
        //batch(); //当消息数量为: 1000 ,批量确认大小为: 100 时,批量确认耗时: 2189
        //3.异步批量确认: 最佳性能和资源使用,在出现错误的情况下可以很好地控制.
        async(); //当消息数量为: 1000 时,异步批量确认耗时: 47

    }

    //单个确认
    public static void single() throws IOException, TimeoutException, InterruptedException {
        String queue_name = "single";
        Channel channel = RabbitMqUtil.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //创建队列
        channel.queueDeclare(queue_name,true,false,false,null);
        long start = System.currentTimeMillis();
        for(int i = 1; i <= NUM; i++){
            String message = String.valueOf(i);
            //发送消息
            channel.basicPublish("",queue_name,null,message.getBytes(StandardCharsets.UTF_8));
            //单个消息就进行发布确认
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println(message+":消息发送成功");
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("当消息数量为: "+NUM+" 时,单个确认耗时: "+(end-start));
    }

    //批量确认
    public static void batch() throws IOException, TimeoutException, InterruptedException {
        String queue_name = "batch";
        Channel channel = RabbitMqUtil.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //创建队列
        channel.queueDeclare(queue_name,true,false,false,null);
        long start = System.currentTimeMillis();
        //批量确认消息大小
        int batchSize = 100;

        for(int i = 1; i <= NUM; i++){
            String message = String.valueOf(i);
            //发送消息
            channel.basicPublish("",queue_name,null,message.getBytes(StandardCharsets.UTF_8));
            if(i % batchSize == 0){
                //进行发布确认
                boolean flag = channel.waitForConfirms();
                if(flag){
                    System.out.println(message+":消息发送成功");
                }
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("当消息数量为: "+NUM+" ,批量确认大小为: "+batchSize+" 时,批量确认耗时: "+(end-start));
    }

    //异步批量确认
    public static void async() throws IOException, TimeoutException, InterruptedException {
        String queue_name = "async";
        Channel channel = RabbitMqUtil.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //创建队列
        channel.queueDeclare(queue_name,true,false,false,null);
        /**
         * 线程安全有序的一个哈希表 适用于高并发的情况下
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序号
         * 3.支持高并发(多线程)
         */
        ConcurrentSkipListMap<Long,String> confirmMap = new ConcurrentSkipListMap<>();
        /**
         * 消息成功的回调函数
         * 1.消息的标识
         * 2.multiple为了支持批量确认模式而提供的。当值为 false 时，表示只确认当前的 deliveryTag 值所对应的单个消息，
         * 当值为 true 时，表示确认所有小于等于当前 deliveryTag 值所对应的消息。
         */
        ConfirmCallback success = (long deliveryTag , boolean multiple ) -> {
            //删除掉已经确认的消息
            //如果是批量的,删除头部的全部
            String message = confirmMap.get(deliveryTag);
            if(multiple){
                ConcurrentNavigableMap<Long, String> headMap = confirmMap.headMap(deliveryTag);
                headMap.clear();
            } else {
                confirmMap.remove(deliveryTag);
            }
            System.out.println("消息: "+message+",成功,tag: "+deliveryTag  + "" +multiple );
        };
        ConfirmCallback fail = (long deliveryTag , boolean multiple ) -> {
            String message = confirmMap.get(deliveryTag);
            System.out.println("未确认的消息: " + message);
            System.out.println("消息失败tag: "+deliveryTag  + "" +multiple );
        };
        //准备消息的监听器,监听哪些消息成功了,哪些失败了
        channel.addConfirmListener(success,fail);
        long start = System.currentTimeMillis();
        for(int i = 1; i <= NUM; i++){
            String message = String.valueOf(i);
            //记录所有要发送的消息
            confirmMap.put(channel.getNextPublishSeqNo(),message);
            //发送消息
            channel.basicPublish("",queue_name,null,message.getBytes(StandardCharsets.UTF_8));
        }
        long end = System.currentTimeMillis();
        System.out.println("当消息数量为: "+NUM+" 时,异步批量确认耗时: "+(end-start));
    }
}
